package com.gzdh.tang.config;

import com.gzdh.tang.comm.util.RedisUtil;
import com.gzdh.tang.dishrecognition.listener.ConsumeListener1;
import com.gzdh.tang.modu.service.ReserveService;
import org.springframework.context.annotation.Configuration;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.data.redis.RedisSystemException;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;

import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import io.lettuce.core.RedisBusyException;
import io.lettuce.core.RedisCommandExecutionException;

import org.springframework.data.redis.connection.stream.*;

import java.time.Duration;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

@Configuration
public class RedisConf {



    @Autowired
    ConsumeListener1 consumeListener1;

    @Autowired
    private RedisUtil redisUtil;

    // @Bean
    // RedisMessageListenerContainer redisContainer(RedisConnectionFactory
    // connectionFactory) {
    // RedisMessageListenerContainer container = new
    // RedisMessageListenerContainer();
    // container.setConnectionFactory(connectionFactory);
    // container.addMessageListener(new WxOrdersListener(reserveService), new
    // ChannelTopic("tenmao.blog.channel"));
    // return container;
    // }

    @Autowired
    private RedisConnectionFactory redisConnectionFactory;

    @Bean(initMethod = "start", destroyMethod = "stop")
    public StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer() {
        System.out.println("初始化");
        AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(), r -> {
                    Thread thread = new Thread(r);
                    thread.setName("async-stream-consumer-" + index.getAndIncrement());
                    thread.setDaemon(true);
                    return thread;
                });

        StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> options =
                StreamMessageListenerContainer.StreamMessageListenerContainerOptions
                .builder()
                // 一次最多获取多少条消息
                .batchSize(3)
                // 运行 Stream 的 poll task
                .executor(executor)
                // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小
                .pollTimeout(Duration.ofSeconds(3))
                // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
                .errorHandler(new StreamErrorHandler())
                .build();

        StreamMessageListenerContainer<String, MapRecord<String, String, String>> streamMessageListenerContainer = StreamMessageListenerContainer
                .create(redisConnectionFactory, options);

        // 独立消费
        String streamKey = "redis-stream-01";

        try {
            redisUtil.createGroup(streamKey, "group-a");
        } catch (RedisSystemException e) {
            if (e.getRootCause().getClass().equals(RedisBusyException.class)) {

            } else if (e.getRootCause().getClass().equals(RedisCommandExecutionException.class)) {
                // log.info("STREAM - Stream does not yet exist, creating empty stream: event-stream");
                // TODO: There has to be a better way to create a stream than this!?
                // redisTemplate.opsForStream().add("event-stream", Collections.singletonMap("", ""));
                // redisTemplate.opsForStream().createGroup("event-stream", "my-group-2");
            } else
                throw e;
        }

        // 消费组A,不自动ack
        // 从消费组中没有分配给消费者的消息开始消费
        streamMessageListenerContainer.receive(Consumer.from("group-a", "consumer-a"),
                StreamOffset.create(streamKey, ReadOffset.lastConsumed()), consumeListener1);

        return streamMessageListenerContainer;
    }

}
